package cn.cmft.jc;

import cn.cmft.jc.config.FlinkConfigReader;
import cn.cmft.jc.sink.WithDorisSink;
import cn.cmft.jc.source.MonitorMySQLSource;
import cn.cmft.jc.transform.DimTableJoinerAnalyze;
import cn.cmft.jc.transform.ExtractFromSchemaChangeData;
import cn.cmft.jc.transform.RecordFormat;
import cn.cmft.jc.transform.joiner.ImpactTaskResult;
import cn.cmft.jc.transform.parse.ChangeOptionMessage;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.text.MessageFormat;
import java.util.List;
import java.util.Optional;


public class TableMonitor {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger(RestOptions.PORT,8081);
        // get env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1).disableOperatorChaining().enableCheckpointing(3000);
//        env.enableCheckpointing(3000, CheckpointingMode.EXACTLY_ONCE);
        //get config
        ParameterTool parameterTool = FlinkConfigReader.getParameterTool();

        // source
        MonitorMySQLSource monitorMySQLSource = new MonitorMySQLSource(parameterTool);
        DataStreamSource<String> streamSource = monitorMySQLSource.getMySQLSource(env);


        streamSource.print("====原始数据====>\n");
        SingleOutputStreamOperator<String> record = streamSource
                //提取ddl事件中，属于库表结构变更的相关事件信息（排除索引/约束/创建表等DDL事件）
                .map(new ExtractFromSchemaChangeData(parameterTool))
                .flatMap(new FlatMapFunction<Optional<ChangeOptionMessage>, ChangeOptionMessage>() {
                    @Override
                    public void flatMap(Optional<ChangeOptionMessage> in, Collector<ChangeOptionMessage> out) throws Exception {
                        in.ifPresent(out::collect);
                    }
                })
                // 关联大数据平台元数据表，确认是业务测数据结构的变更是否对大数据平台侧的采集任务有影响
                .map(new DimTableJoinerAnalyze(parameterTool))
                .flatMap(new RecordFormat());

        record.print("===最终输出数据===>\n");
        record.sinkTo(new WithDorisSink().build());

        env.execute(MessageFormat.format("host:{0} port: {1} Table Structure Monitor",
                parameterTool.get("monitor.datasource.hostname"),
                parameterTool.get("monitor.datasource.port")));

    }
}